package zookeeper;

import java.util.concurrent.Callable;

import org.apache.log4j.Logger;
import org.apache.thrift.protocol.TProtocol;

import zookeeper.ThriftConnectionManager.Action;

public class EngineClientProvider {
	private static final String ZK_ADDTRESS = "103.29.134.218:2181,103.29.134.219:2181,103.29.134.220:2181";
	protected static final String AUTOPACK_PATH = "/service/autopack";
	private static UserObserve uo = new UserObserve();
	private static ThriftConnectionManager connectionManager;
	private static ZK zk;

	private static int timeout;
	private static String zookeeper_address;
	public static Logger logger = Logger.getLogger("push");
	static {
		connectionManager = new ThriftConnectionManager();
		zookeeper_address = ZK_ADDTRESS;
		timeout = 5000;
		try {
			zk.setCallback(new Callable() {
				public Object call() throws Exception {
		                zk.observeNonLeafNode(AUTOPACK_PATH, uo);
		                return null;
				}}
           );
			zk.connect();
		} catch (Exception e) {
			logger.error("zk init error", e);
		}
	}

	private static void getActionMethod(Action ac) {
		String ip;
		int port = 0;
		String config = "";
		if (ac instanceof UserAction) {
			config = uo.getRandomNode();
		}
		String[] configs = config.split(":");
		ip = configs[0];
		if (configs[1] != null && !"".equals(configs[1].trim()))
			port = Integer.valueOf(configs[1]);

		connectionManager.withConnection(ip, port, timeout, ac);
	}

	public abstract static class UserAction<V> extends
			ThriftConnectionManager.Action<UserEngine.Client, V> {
		@Override
		public UserEngine.Client getClient(TProtocol protocol) {
			return new UserEngine.Client(protocol);
		}
	}

	public static <V> V withConnectionUser(UserAction<V> action) {
		getActionMethod(action);
		return action.getVal();
	}
}